package com.xian.java.batch;

import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.io.jdbc.JDBCInputFormat;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.types.Row;
import org.apache.maven.plugins.shade.resource.AppendingTransformer;

/**
 * flink获取jdbc mysql数据
 */
public class GetMysqlData {
    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        //执行sql
        String sql = "SELECT id,NAME,birthday FROM testdb.person";
        System.out.println("待执行sql为:"+sql);

        JDBCInputFormat.JDBCInputFormatBuilder jdbcInputFormatBuilder = JDBCInputFormat.buildJDBCInputFormat();
        jdbcInputFormatBuilder.setDrivername("com.mysql.jdbc.Driver");
        jdbcInputFormatBuilder.setDBUrl("jdbc:mysql://r71/db1");
        jdbcInputFormatBuilder.setUsername("root");
        jdbcInputFormatBuilder.setPassword("123456");
        jdbcInputFormatBuilder.setQuery(sql);
        jdbcInputFormatBuilder.setRowTypeInfo(new RowTypeInfo(
                BasicTypeInfo.INT_TYPE_INFO,
                BasicTypeInfo.STRING_TYPE_INFO,
                BasicTypeInfo.DATE_TYPE_INFO
        ));

        JDBCInputFormat jdbcInputFormat = jdbcInputFormatBuilder.finish();

        DataSource<Row> dataSource = env.createInput(jdbcInputFormat);


        dataSource.print();


        System.out.println("程序运行完毕!");




    }
}
